package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.{ConnUtil, MQObject}
import com.icarbonx.dap.notification.util.{DateUtil, KafkaUtil, NoticeMsg, SparkUtil}
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
 * Created by Lenovo on 2018/8/12.
 * 16.多发运动后低糖警报：连续两个点血糖<3.0mmol/L,5天内出现3次&运动后1小时内：low_glycemia_step_freq：每次运动后：触发及时发送：每小时提示不超过1条
 * 28.运动后低糖：连续两个点血糖<2.8.0mmol/L&运动后1小时内：low_glycemia_step：每次运动后：触发及时发送
 * 总计2个
 */
object BloodSugarSportStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    /**
     * brokers = "10.47.250.4:9092,10.47.250.7:9092,10.47.250.6:9092"
     * topicList = "abel_test"
     * mq:10.104.103.7 5672 guest icx@betaOn16 /
     * redis:10.104.103.7 6379 passxx#$#dslllfjshwerw123@ 3000 abel_test
     */
    println(args.mkString("\t"))
    val Array(brokers, topicList, otherTopicList, mqHost, mqPort, mqUsername,
    mqPassword, mqVirtualHost, exchangeName, mqRouteKey, checkpointDir, duation, rddDuration, appName) = args
    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord], classOf[SportRecord]))

    val sc = SparkContext.getOrCreate(conf)
    //    val checkpointDir =
    //      "/data/streaming/checkpoint/bloodsugar_sport_stream/"

    val updateSugarFunc = (values: Seq[ArrayBuffer[BloodSugarRecord]],
                           state: Option[ArrayBuffer[BloodSugarRecord]]) => {
      //保存状态
      val HourTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 12 * 60 * 60 * 1000
      val currentCount = ArrayBuffer[BloodSugarRecord]()
      values.foreach(list => currentCount ++= list)
      val previousCount = state.getOrElse(ArrayBuffer[BloodSugarRecord]())
      Some(currentCount ++ previousCount.filter(_.timestamp > HourTimeStamp))
    }
    val updateSportFunc = (values: Seq[ArrayBuffer[SportRecord]],
                           state: Option[ArrayBuffer[SportRecord]]) => {
      //保存状态
      val HourTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 12 * 60 * 60 * 1000
      val currentCount = ArrayBuffer[SportRecord]()
      values.foreach(list => currentCount ++= list)
      val previousCount = state.getOrElse(ArrayBuffer[SportRecord]())
      Some(currentCount ++ previousCount.filter(_.endTime > HourTimeStamp))
    }

    val updateSportTimeStateFunc =
      (values: Seq[mutable.HashMap[Long, Int]],
       state: Option[mutable.HashMap[Long, Int]]) => {
        //保存状态
        val fiveDaysAgoTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 5 * 24 * 60 * 60 * 1000
        val currentState = mutable.HashMap[Long, Int]()
        val previousState = state.getOrElse(mutable.HashMap[Long, Int]())
        values.foreach {
          case map =>
            map.foreach {
              case (sportTime, flag) =>
                if ((previousState.contains(sportTime) && previousState(sportTime) == 1)
                  || (currentState.contains(sportTime) && currentState(sportTime) == 1)
                ) {
                  currentState += ((sportTime, 1))
                } else {
                  currentState += ((sportTime, flag))
                }
            }
        }
        val tempOldState = previousState.filter {
          case (sportTime, flag) =>
            !currentState.contains(sportTime) && sportTime > fiveDaysAgoTimeStamp
        }
        Some(currentState ++= tempOldState)
      }

    def creatingFunc(): StreamingContext = {
      val ssc = new StreamingContext(sc, Seconds(duation.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      val mqConn =
        new MQObject(
          mqHost,
          mqPort.toInt,
          mqUsername,
          mqPassword,
          mqVirtualHost
        )

      val sugarTopics = topicList.split(",").toSet
      val sportTopics = otherTopicList.split(",") toSet

      val sugarMessages = KafkaUtils
        .createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          kafkaParams,
          sugarTopics
        )
      val sportMessages = KafkaUtils
        .createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          kafkaParams,
          sportTopics
        )

      val sugarBatch: DStream[(String, ArrayBuffer[BloodSugarRecord])] =
        sugarMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg._2)
              val gson = new Gson()
              val voList =
                gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
              voList.map { vo =>
                (vo.personId.toString, ArrayBuffer[BloodSugarRecord](vo))
              }.toList
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", ArrayBuffer[BloodSugarRecord]()))
            }
          }
          .filter(_._1.nonEmpty)
          .updateStateByKey(updateSugarFunc)

      sugarBatch.checkpoint(Duration(rddDuration.toInt * 1000))

      val sportBatch: DStream[(String, ArrayBuffer[SportRecord])] =
        sportMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg)
              val gson = new Gson()
              val voList = gson.fromJson(msg._2, classOf[Array[SportRecord]])
              val result = ArrayBuffer[(String, ArrayBuffer[SportRecord])]()
              voList.foreach { vo =>
                if (vo.`type` == 2) {
                  result += (
                    (
                      vo.personId.toString,
                      ArrayBuffer[SportRecord](vo)
                      )
                    )
                }
              }
              result
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", ArrayBuffer[SportRecord]()))
            }
          }
          .filter(_._1.nonEmpty)
          .updateStateByKey(updateSportFunc)

      sportBatch.checkpoint(Duration(rddDuration.toInt * 1000))

      val sugarAndSport = sportBatch.join(sugarBatch)
      val sortRdd = sugarAndSport
        .filter {
          case (pid, (sportList, sugarList)) =>
            sportList.nonEmpty && sugarList.nonEmpty
        }
        .map {
          case (pid, (sportList, sugarList)) =>
            (
              pid,
              (sportList.sortBy(_.endTime), sugarList.sortBy(_.timestamp))
              )
        }

      //@todo （多发运动后低糖记录）
      val sportTimeDoneFlag = sortRdd
        .filter(_._2._1.nonEmpty)
        .map {
          case (pid, (sportList, sugarList)) =>
            val result = mutable.HashMap[Long, Int]()
            sportList.foreach { sportVo =>
              result += ((sportVo.endTime, 0))
              val afterSportOneHour = sportVo.endTime + 60 * 60 * 1000
              val afterRangeTimeSugar = sugarList.filter(
                sugarVo =>
                  sugarVo.timestamp >= sportVo.endTime && sugarVo.timestamp < afterSportOneHour
              )
              if (afterRangeTimeSugar.size >= 2) {
                var preVo = afterRangeTimeSugar.head
                for (i <- 1 to afterRangeTimeSugar.size - 1) {
                  val nowVo = afterRangeTimeSugar(i)
                  if (preVo.value < 3 && nowVo.value < 3 &&
                    nowVo.timestamp - preVo.timestamp < 15 * 60 * 60 * 1000
                  ) {
                    result += ((sportVo.endTime, 1))
                  }
                  preVo = nowVo
                }
              }
            }
            (pid, result)
        }
        .updateStateByKey(updateSportTimeStateFunc)

      sportTimeDoneFlag.checkpoint(Duration(rddDuration.toInt * 1000))

      //@todo 16.多发运动后低糖警报：连续两个点血糖<3.0mmol/L,5天内出现3次&运动后1小时内：low_glycemia_step_freq：每次运动后：触发及时发送：每小时提示不超过1条
      sportTimeDoneFlag.foreachRDD { mess =>
        if (!mess.isEmpty) {
          mess.foreachPartition { lines =>
            val mqMsg = ArrayBuffer[String]()
            val gson = new Gson()
            lines.foreach {
              case (pid, sportListState) =>
                println("____ pid :" + pid + "____ sportListState :" + sportListState.toString() + "___ sportListState.values.sum " + sportListState.values.sum)
                if (sportListState.values.sum >= 3) {
                  val msg = new WarningMsg(
                    "glycemia",
                    pid.toLong,
                    NoticeMsg.low_glycemia_step_freq._1,
                    NoticeMsg.low_glycemia_step_freq._2,
                    NoticeMsg.low_glycemia_step_freq._3,
                    sportListState.keys.max.toString,
                    NoticeMsg.low_glycemia_step_freq._5,
                    NoticeMsg.low_glycemia_step_freq._4,
                    NoticeMsg.low_glycemia_step_freq._6
                  )
                  mqMsg += gson.toJson(msg)
                }
            }
            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }
          }
        }
      }

      sortRdd.foreachRDD(mess =>
        if (!mess.isEmpty) {
          mess.foreachPartition(lines => {
            val mqMsg = ArrayBuffer[String]()
            val gson = new Gson()
            lines.foreach {
              case (pid, (sportList, sugarList)) => {
                println("____ pid :" + pid + "\tsportList:" + sportList.map(gson.toJson(_)).toString()
                  + "\t ____ sugarList" + sugarList.map(gson.toJson(_)).toString())

                //@todo 28.运动后低糖：连续两个点血糖<2.8.0mmol/L&运动后1小时内：low_glycemia_step：每次运动后：触发及时发送
                sportList.foreach {
                  sportVo =>
                    val afterSportOneHour = sportVo.endTime + 60 * 60 * 1000
                    val rangeTimeSugar = sugarList.filter(
                      sugarVo =>
                        sugarVo.timestamp >= sportVo.endTime && sugarVo.timestamp <= afterSportOneHour
                    )
                    if (rangeTimeSugar.size >= 2) {
                      var preVo = rangeTimeSugar.head
                      for (i <- 1 to rangeTimeSugar.size - 1) {
                        val nowVo = rangeTimeSugar(i)
                        if (preVo.value < 2.8 && nowVo.value < 2.8 &&
                          nowVo.timestamp - preVo.timestamp < 15 * 60 * 60 * 1000) {
                          val msg = new WarningMsg(
                            "glycemia",
                            pid.toLong,
                            NoticeMsg.low_glycemia_step._1,
                            NoticeMsg.low_glycemia_step._2,
                            NoticeMsg.low_glycemia_step._3,
                            sportVo.endTime.toString,
                            NoticeMsg.low_glycemia_step._5,
                            NoticeMsg.low_glycemia_step._4,
                            NoticeMsg.low_glycemia_step._6
                          )
                          mqMsg += gson.toJson(msg)
                        }
                        preVo = nowVo
                      }
                    }
                }
              }
            }
            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }
          })
        })
      ssc
    }



    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }
}